package org.com.blbl.PartitionPer;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

public class PartitionPerReducer extends Reducer<Text, LongWritable, Text, Text>  {

    Integer totalCount = 0;
    private Map<String, Integer> countMap = new HashMap<>();


    @Override
    protected void reduce(Text key, Iterable<LongWritable> values, Reducer<Text, LongWritable, Text, Text>.Context context) throws IOException, InterruptedException {
        //定义次数结果
        IntWritable result = new IntWritable();
        //统计每个分区变量
        int count = 0;
        //遍历一组数组，统计次数，取出该组的所有的Value
        for (LongWritable value : values){
            //所有的value进行累加，就是该单词出现的总次数
            count += value.get();
        }
        result.set(count);
        String key1 = new String(String.valueOf(key));
        Integer result1 = new Integer(result.get());
        totalCount += result1;
        countMap.put(key1, result1);
    }

    @Override
    protected void cleanup(Reducer<Text, LongWritable, Text, Text>.Context context) throws IOException, InterruptedException {
        for (Map.Entry<String, Integer> entry : countMap.entrySet()){
            String partition = entry.getKey();
            int count = entry.getValue();
            double percentage = (double) count / totalCount * 100;
            context.write(new Text(partition), new Text( String.format("%.2f", percentage) + "%"));
        }
    }
}
